package com.jsy.mqhelper.charlotterpc.reference.mq.rabbit;

import com.jsy.mqhelper.charlotterpc.config.CharlotteProperties;
import com.jsy.mqhelper.charlotterpc.config.CharlottePropertyManager;
import com.jsy.mqhelper.charlotterpc.core.mq.rabbit.RabbitConnectionFactory;
import com.jsy.mqhelper.charlotterpc.exception.RemoteEndpointException;
import com.jsy.mqhelper.charlotterpc.exception.RpcTimeoutException;
import com.jsy.mqhelper.charlotterpc.protocol.CharlotteRpcRequest;
import com.jsy.mqhelper.charlotterpc.protocol.CharlotteRpcResponse;
import com.jsy.mqhelper.charlotterpc.util.CharlotteRpcUtil;
import com.jsy.mqhelper.common.core.MetaFunction;
import com.jsy.mqhelper.common.core.MetaInterface;
import com.jsy.mqhelper.common.exception.MetaFunctionDefineException;
import com.jsy.mqhelper.common.util.JsonUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import static com.jsy.mqhelper.common.constant.Constants.MQ_HELPER;

/**
 * @author: SongyangJi
 * @description:
 * @since: 2021/9/28
 */
@Slf4j(topic = MQ_HELPER)
public class RabbitMQProxyInvocationHandler implements InvocationHandler {

    RabbitConnectionFactory rabbitConnectionFactory;

    public RabbitMQProxyInvocationHandler(RabbitConnectionFactory rabbitConnectionFactory) {
        this.rabbitConnectionFactory = rabbitConnectionFactory;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        Class<?> interfaceClass = method.getDeclaringClass();
        if (!interfaceClass.isInterface()) {
            throw new MetaFunctionDefineException(String.format("method:%s not declared in interface", method));
        }

        MetaInterface metaInterface = new MetaInterface(interfaceClass);
        String rpcQueueName = CharlotteRpcUtil.generateQueueName(metaInterface);


        try (Channel channel = rabbitConnectionFactory.createConnection().createChannel()) {
            CharlotteRpcRequest rpcRequest = new CharlotteRpcRequest(MetaFunction.getFunctionId(method), args);

            /* not a good practice to use an anonymous queue*/
            String replyQueueName = channel.queueDeclare().getQueue();
            AMQP.BasicProperties props = new AMQP.BasicProperties
                    .Builder()
                    .replyTo(replyQueueName)
                    .build();

            channel.basicPublish("", rpcQueueName, props, JsonUtil.stringify(rpcRequest).getBytes(StandardCharsets.UTF_8));

            BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

            String cTag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) ->
                    response.offer(new String(delivery.getBody(), StandardCharsets.UTF_8)), consumerTag -> {
            });


            CharlotteProperties charlotteProperties = CharlottePropertyManager.getCharlotteProperties();
            log.debug("charlotteProperties={}", charlotteProperties);

            String rpcResponseStr = response.poll(charlotteProperties.getRpcTimeout(), TimeUnit.SECONDS);
            if(rpcResponseStr == null) {
                throw new RpcTimeoutException();
            }
            channel.basicCancel(cTag);

            CharlotteRpcResponse rpcResponse = JsonUtil.parse(rpcResponseStr, CharlotteRpcResponse.class);
            if (rpcResponse.getError() != null) {
                log.error("remote service endpoint process rpc request error={}", rpcResponse.getError());
                throw new RemoteEndpointException();
            } else if (rpcResponse.getException() != null) {
                throw rpcResponse.getException();
            }
            log.debug("rpcResponse={}", rpcResponse);
            log.debug("rpc.result={}, rpc.resultClass={}", rpcResponse.getResult(), rpcResponse.getResult().getClass());
            return rpcResponse.getResult();
        }
    }

    @PostConstruct
    public void init() {
        log.info("RabbitMQProxyInvocationHandler create successfully.");
    }

}
